package com.shujia.flink.sink;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

public class Demo4MySink {
    public static void main(String[] args)throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        /*
         * addSink: 类似spark中的foreach
         */
        //使用自定义sink
        linesDS.addSink(new MySink());

        env.execute();

    }

    static class MySink implements SinkFunction<String> {
        /**
         * invoke方法每一条数据执行一次
         *
         * @param value   一行数据
         * @param context 上下文对象
         */
        @Override
        public void invoke(String value, Context context) throws Exception {
            System.out.println("自定义sink:" + value);
        }
    }
}
